package com.gxl.butt.domainkafka;

import java.util.Properties;

import com.gxl.butt.dto.bo.Topic;
import org.apache.kafka.common.security.JaasUtils;
import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;
import org.springframework.kafka.KafkaException;

public class KafkaUtil {
    
     public static void createKafaTopic(String ZkStr,Topic topic) throws KafkaException {

         ZkUtils zkUtils = ZkUtils.apply(ZkStr, 30000, 30000,JaasUtils.isZkSecurityEnabled());
         
//        AdminUtils.createTopic(zkUtils, topic.getTopicname(),topic.getPartitionsum(),topic.getReplicationsum(), new Properties(), new RackAwareMode.Enforced$());
         AdminUtils.createTopic(zkUtils, topic.getTopicname(), topic.getPartitionsum(), topic.getReplicationsum(), new Properties(), RackAwareMode.Enforced$.MODULE$);
        zkUtils.close();

    }
     
     public static void deleteKafaTopic(String ZkStr,Topic topic) {

         ZkUtils zkUtils = ZkUtils.apply(ZkStr, 30000, 30000,JaasUtils.isZkSecurityEnabled());
         
        AdminUtils.deleteTopic(zkUtils, topic.getTopicname());

        zkUtils.close();

    }

}